
package mvcc

import (
	"context"
	"errors"
	"fmt"
	"hash/crc32"
	"math"
	"sync"
	"time"

	"go.etcd.io/etcd/api/v3/mvccpb"
	"go.etcd.io/etcd/pkg/v3/schedule"
	"go.etcd.io/etcd/pkg/v3/traceutil"
	"go.etcd.io/etcd/server/v3/lease"
	"go.etcd.io/etcd/server/v3/storage/backend"
	"go.etcd.io/etcd/server/v3/storage/schema"

	"go.uber.org/zap"
)

var (
	ErrCompacted = errors.New("mvcc: required revision has been compacted")
	ErrFutureRev = errors.New("mvcc: required revision is a future revision")
)

const (
	// markedRevBytesLen是已标记修订的字节长度。
	// 第一个'revBytesLen'字节表示正常修订。最后一个
	// 一个字节是标记。
	markedRevBytesLen      = revBytesLen + 1
	markBytePosition       = markedRevBytesLen - 1
	markTombstone     byte = 't'
)

var restoreChunkKeys = 10000 // 用于测试的非常量
var defaultCompactBatchLimit = 1000
var minimumBatchInterval = 10 * time.Millisecond

type StoreConfig struct {
	CompactionBatchLimit    int
	CompactionSleepInterval time.Duration
}

type store struct {
	ReadView
	WriteView

	cfg StoreConfig

	// 在开启只读／读写事务时，需要获取该读锁进行同步，
	// 即在Read（）方法和Write（）方法中获取该读锁，在End（）方法中释放。
	// 在进行压缩等非事务性的操作时，需要加写锁进行同步。
	mu sync.RWMutex
	// 当前store实例关联的后端存储，
	b       backend.Backend
	// 当前store实例关联的内存索引，
	kvindex index
	// 租约相关的内容，
	le lease.Lessor

	// 在修改currentRev字段和compactMainRev字段时，需要获取该锁进行同步。
	revMu sync.RWMutex
	// 该字段记录当前的revision信息（mainrevision部分的值）。
	currentRev int64
	// 该字段记录最近一次压缩后最小的revision信息（mainrevision部分的值〉。
	compactMainRev int64
	// FIFO调度器，
	fifoSched schedule.Scheduler

	stopc chan struct{}

	lg *zap.Logger
}

// NewStore（）函数，其中完成了store实例的初始化过程
func NewStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfig) *store {
	if lg == nil {
		lg = zap.NewNop()
	}
	if cfg.CompactionBatchLimit == 0 {
		cfg.CompactionBatchLimit = defaultCompactBatchLimit
	}
	if cfg.CompactionSleepInterval == 0 {
		cfg.CompactionSleepInterval = minimumBatchInterval
	}
	s := &store{
		cfg:     cfg,
		b:       b,	// 初始化backend字段
		kvindex: newTreeIndex(lg),	// 初始化kvindex字段

		le: le,

		currentRev:     1,	// currentRev字段初始化为1
		compactMainRev: -1,	// compactMainRev字段初始化为－1

		fifoSched: schedule.NewFIFOScheduler(),

		stopc: make(chan struct{}),

		lg: lg,
	}
	// 创建readV工ew实例和writeView实例
	s.ReadView = &readView{s}
	s.WriteView = &writeView{s}
	if s.le != nil {
		s.le.SetRangeDeleter(func() lease.TxnDelete { return s.Write(traceutil.TODO()) })
	}
	// 获取backend的读写事务，创建名为“key”和“meta”的两个Bucket，然后提交事务
	tx := s.b.BatchTx()
	tx.Lock()
	tx.UnsafeCreateBucket(schema.Key)
	schema.UnsafeCreateMetaBucket(tx)
	tx.Unlock()
	s.b.ForceCommit()

	s.mu.Lock()
	defer s.mu.Unlock()
	// 从backend中恢复当前store的所有状态，其中包括内存中的BTree索引等，
	if err := s.restore(); err != nil {
		// TODO:在此处返回错误而不是恐慌？
		panic("failed to recover store from backend")
	}

	return s
}

func (s *store) compactBarrier(ctx context.Context, ch chan struct{}) {
	if ctx == nil || ctx.Err() != nil {
		select {
		case <-s.stopc:
		default:
			// 修复mvcc中的死锁，有关更多信息，请参阅pr 11817。
			// s.stopc仅在还原操作中更新，该操作由应用
			// 快照调用调用，压缩和应用快照请求由
			// raft序列化，不会同时发生。
			s.mu.Lock()
			f := func(ctx context.Context) { s.compactBarrier(ctx, ch) }
			s.fifoSched.Schedule(f)
			s.mu.Unlock()
		}
		return
	}
	close(ch)
}

func (s *store) Hash() (hash uint32, revision int64, err error) {
	// TODO:哈希和修订可能不一致，一个可能的修复方法是添加s.revMu。函数开始处的RLock（），成本高昂
	start := time.Now()

	s.b.ForceCommit()
	h, err := s.b.Hash(schema.DefaultIgnores)

	hashSec.Observe(time.Since(start).Seconds())
	return h, s.currentRev, err
}

func (s *store) HashByRev(rev int64) (hash uint32, currentRev int64, compactRev int64, err error) {
	start := time.Now()

	s.mu.RLock()
	s.revMu.RLock()
	compactRev, currentRev = s.compactMainRev, s.currentRev
	s.revMu.RUnlock()

	if rev > 0 && rev <= compactRev {
		s.mu.RUnlock()
		return 0, 0, compactRev, ErrCompacted
	} else if rev > 0 && rev > currentRev {
		s.mu.RUnlock()
		return 0, currentRev, 0, ErrFutureRev
	}

	if rev == 0 {
		rev = currentRev
	}
	keep := s.kvindex.Keep(rev)

	tx := s.b.ReadTx()
	tx.RLock()
	defer tx.RUnlock()
	s.mu.RUnlock()

	upper := revision{main: rev + 1}
	lower := revision{main: compactRev + 1}
	h := crc32.New(crc32.MakeTable(crc32.Castagnoli))

	h.Write(schema.Key.Name())
	err = tx.UnsafeForEach(schema.Key, func(k, v []byte) error {
		kr := bytesToRev(k)
		if !upper.GreaterThan(kr) {
			return nil
		}
		// 跳过计划删除的修订
		// 由于压缩；如果没有，就不要跳过。
		if lower.GreaterThan(kr) && len(keep) > 0 {
			if _, ok := keep[kr]; !ok {
				return nil
			}
		}
		h.Write(k)
		h.Write(v)
		return nil
	})
	hash = h.Sum32()

	hashRevSec.Observe(time.Since(start).Seconds())
	return hash, currentRev, compactRev, err
}

func (s *store) updateCompactRev(rev int64) (<-chan struct{}, error) {
	s.revMu.Lock()
	if rev <= s.compactMainRev {
		ch := make(chan struct{})
		f := func(ctx context.Context) { s.compactBarrier(ctx, ch) }
		s.fifoSched.Schedule(f)
		s.revMu.Unlock()
		return ch, ErrCompacted
	}
	if rev > s.currentRev {
		s.revMu.Unlock()
		return nil, ErrFutureRev
	}
	// 更新compactMainRev
	s.compactMainRev = rev

	SetScheduledCompact(s.b.BatchTx(), rev)
	// 提交当前读写事务
	s.b.ForceCommit()

	s.revMu.Unlock()

	return nil, nil
}

func (s *store) compact(trace *traceutil.Trace, rev int64) (<-chan struct{}, error) {
	ch := make(chan struct{})
	var j = func(ctx context.Context) {
		if ctx.Err() != nil {
			s.compactBarrier(ctx, ch)
			return
		}
		start := time.Now()
		keep := s.kvindex.Compact(rev)
		indexCompactionPauseMs.Observe(float64(time.Since(start) / time.Millisecond))
		// 这里是真正对BoltDB进行压缩的地方
		if !s.scheduleCompaction(rev, keep) {
			s.compactBarrier(context.TODO(), ch)
			return
		}
		close(ch)	// 当正常压缩介绍之后，会关闭该通过，通知其他监听的
	}
	// store fifoSched是一个FIFOScheduler，在Scheduler接口中定义了一个Schedule(j Job) 方法，
	// 该方法会将Job放入调度器进行调度，
	// 其中的Job实际上就是一个func(context.Context）函数这里将对BoltDB的压缩操作（即上面定义的〕函数）放入FIFO调度器中，异步执行
	s.fifoSched.Schedule(j)
	trace.Step("schedule compaction")
	return ch, nil
}

func (s *store) compactLockfree(rev int64) (<-chan struct{}, error) {
	ch, err := s.updateCompactRev(rev)
	if err != nil {
		return ch, err
	}

	return s.compact(traceutil.TODO(), rev)
}

// store.Compact（）方法主要用于实现键值对的压缩。该方法首先将传入的参数转换成revision实例，
// 然后在metaBucket中记录此次压缩的键值对信息（Key为scheduledCompactRev, Value为revision），
// 之后调用treeIndex. Compact（）方法完成对内存索引的压缩，最后通过FIFOScheduler异步完成对BoltDB的压缩操作。
func (s *store) Compact(trace *traceutil.Trace, rev int64) (<-chan struct{}, error) {
	s.mu.Lock()

	ch, err := s.updateCompactRev(rev)
	trace.Step("check and update compact revision")
	if err != nil {
		s.mu.Unlock()
		return ch, err
	}
	s.mu.Unlock()

	return s.compact(trace, rev)
}

func (s *store) Commit() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.b.ForceCommit()
}

// Raft协议，Follower节点在收到快照数据时，会使用快照数据恢复当前节点的状态。
// 在这个恢复过程中就会调用store.Restore（）方法完成内存索引和store中其他状态的恢复，
func (s *store) Restore(b backend.Backend) error {
	// 获取mu上的写锁，关闭当前的FIFOScheduler
	s.mu.Lock()
	defer s.mu.Unlock()
	close(s.stopc)
	s.fifoSched.Stop()
	// 更新store.b字段，指向新的backend实例
	s.b = b
	// 更新store.kvindex字段，指向新的内存索引
	s.kvindex = newTreeIndex(s.lg)

	{
		// 在恢复期间，如果总的挂起密钥超过恢复
		s.revMu.Lock()
		// 重置currentRev字段和compactMainRev字段
		s.currentRev = 1
		s.compactMainRev = -1
		s.revMu.Unlock()
	}
	// 更新fifoSched字段，指向新的FIFOII Scheduler
	s.fifoSched = schedule.NewFIFOScheduler()
	s.stopc = make(chan struct{})
	// 调用restore（）开始的恢复内存索引
	return s.restore()
}

// store.restore（）方法恢复内存索引的大致逻辑是：首先批量读取BoltDB中所有的键值对数据，
// 然后将每个键值对封装成revKey Value实例，并写入一个通道中，
// 最后由另一个单独的goroutine读取该通道，并完成内存索引的恢复
// store.restore（）方 法 ，其中除了恢复内存索引，还会重新为键值对绑定相关的 Lease 实例
//  restore （）方法中就会应用 keyToLease 这个 map 中记录的对应关系，完成相应的恢复操作
func (s *store) restore() error {
	s.setupMetricsReporter()
	// 创建min和max，后续在BoltDB中进行范围查询时的起始key和结束key就是min和max
	min, max := newRevBytes(), newRevBytes()
	revToBytes(revision{main: 1}, min)	// min为1_0
	revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max)

	keyToLease := make(map[string]lease.LeaseID)

	// 获取读写事务，并加锁
	tx := s.b.BatchTx()
	tx.Lock()

	finishedCompact, found := UnsafeReadFinishedCompact(tx)
	if found {
		s.revMu.Lock()
		s.compactMainRev = finishedCompact

		s.lg.Info(
			"restored last compact revision",
			zap.String("meta-bucket-name-key", string(schema.FinishedCompactKeyName)),
			zap.Int64("restored-compact-revision", s.compactMainRev),
		)
		s.revMu.Unlock()
	}
	scheduledCompact, _ := UnsafeReadScheduledCompact(tx)
	// 从tx 
	keysGauge.Set(0)
	// 在restorelntolndex（）方法中会启动一个单独的goroutine，用于接收从backend中读取的键值对数据，
	// 并恢复到新建的内存索引中｛store.kv工ndex），
	// 这里需要了解该方法的返回位rkvc是一个通道，也是当前goroutine与上述goroutine通信的桥梁
	rkvc, revc := restoreIntoIndex(s.lg, s.kvindex)
	for {
		// 调用UnsafeRange（）方法查询BoltDB中的keyBucket，返回键位对数量的上限是restoreChunkKeys,默认是10000
		// 查询 restoreChunkKeys 数量的键值对
		keys, vals := tx.UnsafeRange(schema.Key, min, max, int64(restoreChunkKeys))
		if len(keys) == 0 {	// 查询结果为空，则直接结束当前for循环
			break
		}
		// 将查询到的键值对数据写入rkvc这个通道中，并由restorelntolndex（）方法中创建的 goroutine进行处理，
		restoreChunk(s.lg, rkvc, keys, vals, keyToLease)
		if len(keys) < restoreChunkKeys {
			// 范围查询得到的结采数小于restoreChunkKeys，即表示最后一次查询
			break
		}
		// 更新min作为下一次范围查询的起始key
		newMin := bytesToRev(keys[len(keys)-1][:revBytesLen])
		newMin.sub++
		revToBytes(newMin, min)
	}
	close(rkvc)	// 关闭rkvc通道

	{
		s.revMu.Lock()
		// 从revc通道中获取恢复之后的currentRev
		s.currentRev = <-revc

		// 校正currentRev和scheduledCompact
		if s.currentRev < s.compactMainRev {
			s.currentRev = s.compactMainRev
		}
		s.revMu.Unlock()
	}
	if scheduledCompact <= s.compactMainRev {
		scheduledCompact = 0
	}
	// 租期相关的处理，
	for key, lid := range keyToLease {
		if s.le == nil {
			tx.Unlock()
			panic("no lessor to attach lease")
		}
		err := s.le.Attach(lid, []lease.LeaseItem{{Key: key}})	// 绑定键位对与指定的 Lease 实例
		if err != nil {
			s.lg.Error(
				"failed to attach a lease",
				zap.String("lease-id", fmt.Sprintf("%016x", lid)),
				zap.Error(err),
			)
		}
	}

	tx.Unlock()	// 读写事务结束，释放锁

	s.lg.Info("kvstore restored", zap.Int64("current-rev", s.currentRev))
	// 如果在开始恢复之前，存在未执行完的压缩操作，则重启该压缩操作
	if scheduledCompact != 0 {
		if _, err := s.compactLockfree(scheduledCompact); err != nil {
			s.lg.Warn("compaction encountered error", zap.Error(err))
		}

		s.lg.Info(
			"resume scheduled compaction",
			zap.Int64("scheduled-compact-revision", scheduledCompact),
		)
	}

	return nil
}

type revKeyValue struct {
	key  []byte	// BoltDB中的Key值，可以转换得到的revision实例。
	kv   mvccpb.KeyValue	// BoltDB中保存的Value值。
	kstr string	// 原始的Key值。
}

// 启动一个后台的goroutine，读取rkvc通道中的revK巳yValue实例，
// 并将其中的键值对数据’恢复到内存索引中等一系列操作，都是在restorelntolndexO函数中完成的，
func restoreIntoIndex(lg *zap.Logger, idx index) (chan<- revKeyValue, <-chan int64) {
	// 创建两个通道，其中rkvc通道用来传递revKeyValue实例的通道
	rkvc, revc := make(chan revKeyValue, restoreChunkKeys), make(chan int64, 1)
	go func() {	// 启动一个单独的goroutine
		currentRev := int64(1)	// 记录当前遇到的最大的mainrevision位
		// 在该goroutine结束时（即 rkvc通过被关闲时），
		// 将当前已知的最大mainrevis工on值写入 revc通道中，待其他goroutine读取
		defer func() { revc <- currentRev }()
		// 虽然BTree的查找效率很高，但是随着BTree层次的加深，效率也随之下降，这里使用k工Cache这个map做了一层缓存，
		// 更加高效地查找对应的keyindex实例前面从BoltDB中读取到的键值对数据并不是按照原始Key值进行排序的，
		// 如果直接向BTree中写入，则可能会引起节点的分裂等交换操作，效率比较低。所以这里使用kiCache这个map做了一层缓存
		kiCache := make(map[string]*keyIndex, restoreChunkKeys)
		for rkv := range rkvc {	// 从rkvc通道中读取revKeyValue实例
			ki, ok := kiCache[rkv.kstr]		// 先查询缓存
			// 如果kiCache中缓存了大量的Key，但是依然没有命中，则清理缓存
			if !ok && len(kiCache) >= restoreChunkKeys {
				i := 10
				for k := range kiCache {
					delete(kiCache, k)
					if i--; i == 0 {	// 只清理10个Key
						break
					}
				}
			}
			// 如果缓存未命中，则从内存索引中查找对应的keyIndex实例，并添加到缓存中
			if !ok {
				ki = &keyIndex{key: rkv.kv.Key}
				if idxKey := idx.KeyIndex(ki); idxKey != nil {
					kiCache[rkv.kstr], ki = idxKey, idxKey
					ok = true
				}
			}
			// 将revKeyValue.key转换成revision实例
			rev := bytesToRev(rkv.key)	// 更新currentRev位
			currentRev = rev.main
			if ok {
				if isTombstone(rkv.key) {	// 当前revKeyValue实例对应一个tombstone键值对
					if err := ki.tombstone(lg, rev.main, rev.sub); err != nil {
						// 在对应的keyIndex实例中插入tombstone
						lg.Warn("tombstone encountered error", zap.Error(err))
					}
					continue
				}
				// 在对应keyindex实例中添加正常的revis工on信息
				ki.put(lg, rev.main, rev.sub)
			} else if !isTombstone(rkv.key) {
				// 如果从内存索引中依然未查询到对应的keyIndex实例，则需妥填充keyindex实例中其他字段，并添加到内存索引中
				ki.restore(lg, revision{rkv.kv.CreateRevision, 0}, rev, rkv.kv.Version)
				idx.Insert(ki)
				kiCache[rkv.kstr] = ki	// 同时会将该keyIndex实例添加到kiCache缓存中
			}
		}
	}()
	return rkvc, revc
}

// // store.restore（）方法中，通过UnsafeRange（）方法从BoltDB中查询到的键值对，
// // 然后通过r巳storeChunk（）函数转换成revKey Value实例，并写入rkvc通道中。
// restoreChunk（）方法，其中会将键值对和 Lease 实例的绑定关系记录到 keyToLease 这个 map 中
func restoreChunk(lg *zap.Logger, kvc chan<- revKeyValue, keys, vals [][]byte, keyToLease map[string]lease.LeaseID) {
	for i, key := range keys {	// 追历读取到的rev工sion信息
		rkv := revKeyValue{key: key}	// 创建revKeyValue实例
		if err := rkv.kv.Unmarshal(vals[i]); err != nil {	// 反序列化得到KeyValue
			lg.Fatal("failed to unmarshal mvccpb.KeyValue", zap.Error(err))
		}
		}
		rkv.kstr = string(rkv.kv.Key)	// 记录对应的原始Key值
		if isTombstone(key) {	// 如果是 Tombstone ，则从 keyToLease 中删除
			delete(keyToLease, rkv.kstr)	// 删除tombstone标识的键值对
		} else if lid := lease.LeaseID(rkv.kv.Lease); lid != lease.NoLease {
			keyToLease[rkv.kstr] = lid	// 如果键位对有绑定的 Lease 实例，则记录到j keyToLease 中
		} else {
			delete(keyToLease, rkv.kstr)	// 如果该键值对没有绑定 Lease ，则从 keyToLease 中删除
		}
		kvc <- rkv	// 将上述revKeyValue实例写入rkvc逼近中，等待处理
	}
}

func (s *store) Close() error {
	close(s.stopc)
	s.fifoSched.Stop()
	return nil
}

func (s *store) setupMetricsReporter() {
	b := s.b
	reportDbTotalSizeInBytesMu.Lock()
	reportDbTotalSizeInBytes = func() float64 { return float64(b.Size()) }
	reportDbTotalSizeInBytesMu.Unlock()
	reportDbTotalSizeInBytesDebugMu.Lock()
	reportDbTotalSizeInBytesDebug = func() float64 { return float64(b.Size()) }
	reportDbTotalSizeInBytesDebugMu.Unlock()
	reportDbTotalSizeInUseInBytesMu.Lock()
	reportDbTotalSizeInUseInBytes = func() float64 { return float64(b.SizeInUse()) }
	reportDbTotalSizeInUseInBytesMu.Unlock()
	reportDbOpenReadTxNMu.Lock()
	reportDbOpenReadTxN = func() float64 { return float64(b.OpenReadTxN()) }
	reportDbOpenReadTxNMu.Unlock()
	reportCurrentRevMu.Lock()
	reportCurrentRev = func() float64 {
		s.revMu.RLock()
		defer s.revMu.RUnlock()
		return float64(s.currentRev)
	}
	reportCurrentRevMu.Unlock()
	reportCompactRevMu.Lock()
	reportCompactRev = func() float64 {
		s.revMu.RLock()
		defer s.revMu.RUnlock()
		return float64(s.compactMainRev)
	}
	reportCompactRevMu.Unlock()
}

// appendMarkTombstone将逻辑删除标记附加到正常修订字节。
func appendMarkTombstone(lg *zap.Logger, b []byte) []byte {
	if len(b) != revBytesLen {
		lg.Panic(
			"cannot append tombstone mark to non-normal revision bytes",
			zap.Int("expected-revision-bytes-size", revBytesLen),
			zap.Int("given-revision-bytes-size", len(b)),
		)
	}
	return append(b, markTombstone)
}

// isTombstone检查修订字节是否为墓碑。
func isTombstone(b []byte) bool {
	return len(b) == markedRevBytesLen && b[markBytePosition] == markTombstone
}
